package com.xy.job.admin.core

import com.xy.job.admin.common.JOB_DISTRIBUTE_INTERVAL
import com.xy.job.admin.domain.JobLock
import com.xy.job.admin.enums.NodeType
import com.xy.job.admin.repository.JobInfoRepository
import com.xy.job.admin.repository.JobLockRepository
import com.xy.job.admin.repository.NodeRegistryRepository
import com.xy.job.admin.service.NodeRegistryService
import org.slf4j.LoggerFactory
import org.springframework.scheduling.annotation.Scheduled
import org.springframework.stereotype.Component
import java.util.concurrent.TimeUnit

/**
 * master节点负责分发任务给各个调度器节点，每个调度器只调度自己负责的任务
 */
@Component
class JobDistributeExecutor(
    private var inetAddressHolder: InetAddressHolder,
    private var nodeRegistryService: NodeRegistryService,
    private var jobLockRepository: JobLockRepository,
    private var nodeRegistryRepository: NodeRegistryRepository,
    private var jobInfoRepository: JobInfoRepository,
) {

    private val log = LoggerFactory.getLogger(this::class.java)

    @Scheduled(fixedDelay = JOB_DISTRIBUTE_INTERVAL, timeUnit = TimeUnit.SECONDS)
    fun distribute() {
        if (!nodeRegistryService.isAdminMasterNode(inetAddressHolder.address)) {
            log.debug("distribute lock cancel: this node is not master node")
            return
        }
        val schedulerNodes = nodeRegistryRepository.findByNodeType(nodeType = NodeType.ADMIN_SCHEDULER)
        if (schedulerNodes.isEmpty()) {
            log.debug("distribute lock cancel: there is no scheduler node")
            return
        }
        val lockNamesFromEnableJob = jobInfoRepository.findLockNameFromEnableJob()
        if (lockNamesFromEnableJob.isEmpty()) {
            log.debug("distribute lock cancel: there is no enabled job")
            return
        }
        // 只允许分配没有拥有者的锁
        val lockNamesWithoutOwner = jobLockRepository.findLockNameWithoutOwner(lockNames = lockNamesFromEnableJob)
        if (lockNamesWithoutOwner.isEmpty()) {
            return
        }
        val chunkedLockList = lockNamesWithoutOwner.chunked(schedulerNodes.size)
        val jobLockDistributeList = schedulerNodes.zip(chunkedLockList)
            .map { (scheduleNode, lockNames) ->
                lockNames.map { lockName -> JobLock(lockName = lockName, owner = scheduleNode.nodeAddress) }
            }
            .flatten()
        jobLockRepository.saveAllAndFlush(jobLockDistributeList)
    }
}